package com.swak.paxos.statemachine.master;

import java.util.Random;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.paxos.PaxosConfig;
import com.swak.paxos.Value;
import com.swak.paxos.impl.PaxosNode;
import com.swak.paxos.statemachine.StateMachine;
import com.swak.reactivex.threads.EventLoop;
import com.swak.reactivex.threads.SingleThreadEventLoop;
import com.swak.reactivex.threads.SwakThreadFactory;
import com.swak.reactivex.threads.waitstrategy.WakeupWaitStrategy;
import com.swak.timer.RTT;
import com.swak.utils.StringUtils;

import lombok.Getter;

/**
 * MasterStateMachine.
 * 
 * @author 618lf
 */
public class MasterStateMachine implements StateMachine {

	private static final Logger logger = LoggerFactory.getLogger(MasterStateMachine.class);
	private static final EventLoop EventLoop = new SingleThreadEventLoop(
			new SwakThreadFactory("Paxos.Master-ALL", false), new WakeupWaitStrategy());

	/**
	 * 如下三个信息是在 Paxos 算法中会修改此信息, 其他线程会读取此信息
	 */
	@Getter
	private volatile String masterNodeID;
	private volatile long masterVersion;
	private volatile long absExpireTimeMS;

	private volatile boolean needDropMaster;
	private final PaxosNode node;
	private final String nodeID;
	private final int groupID;
	private final EventLoop eventLoop = EventLoop;
	private final long leaseTimeMS;
	private AtomicBoolean running = new AtomicBoolean(false);

	private final Runnable beMasterTask = this::tryBeMasterTask;
	private final RTT rttCal = new RTT();
	private long runTimeAVG = 100;
	private final Random random = new Random();

	public MasterStateMachine(int groupID, PaxosConfig config, PaxosNode node) {
		this.node = node;
		this.nodeID = node.id();
		this.groupID = groupID;
		this.leaseTimeMS = config.getMasterLeaseTimeMS();
	}

	/**
	 * 如果还没有 Master 节点，则发起成为 Master 节点
	 */
	public void checkBeMaster() {

		if (StringUtils.isNotBlank(this.masterNodeID) || this.running.get()) {
			return;
		}

		if (logger.isDebugEnabled()) {
			logger.debug("Start try Be Master");
		}

		this.eventLoop.execute(beMasterTask);
	}

	private void tryBeMasterTask() {

		/*
		 * 设置为运行中
		 */
		if (!(running.compareAndExchange(false, true) == false)) {
			return;
		}

		try {

			/*
			 * Lease Time
			 */
			long nLeaseTime = this.leaseTimeMS;
			long beginTime = System.currentTimeMillis();

			/*
			 * Sync Be Master
			 */
			this.tryBeMaster(nLeaseTime);

			/*
			 * End Be Master
			 */
			long endTime = System.currentTimeMillis();
			long runTime = endTime > beginTime ? (endTime - beginTime) : 0;
			this.runTimeAVG = this.rttCal.calculate(runTime).getEstimateRTT();

			/*
			 * Next Lease Time;
			 * 算法： nLeaseTime - runTime 即可, 一定要少于这个时间
			 */
			long continueLeaseTimeout = (nLeaseTime - this.runTimeAVG) / 2;
			continueLeaseTimeout = (long) ((random.nextDouble() * this.leaseTimeMS) % continueLeaseTimeout
					+ continueLeaseTimeout);

			/*
			 * 重新选取 Master
			 */
			if (this.needDropMaster) {
				this.needDropMaster = false;
				continueLeaseTimeout = nLeaseTime * 2;
			}

			/*
			 * 定时任务的时间，主要是起来检查是否需要发起续约；
			 * 太长：导致 Master 不稳定;
			 * 太短：频繁的发起续租
			 */
			long needSleepTime = continueLeaseTimeout > this.runTimeAVG ? (continueLeaseTimeout - this.runTimeAVG) : 0;

			/*
			 * 提交一次任务
			 */
			this.eventLoop.schedule(this.beMasterTask, needSleepTime, TimeUnit.MILLISECONDS);

			/*
			 * 下一次发起 Try Be Master 的 时间
			 */
			if (logger.isDebugEnabled()) {
				logger.debug("Next Try Be Master, Time:{}", needSleepTime);
			}

		} finally {
			this.running.set(false);
		}
	}

	private void tryBeMaster(long leaseTime) {

		if (logger.isDebugEnabled()) {
			logger.debug("Try Be Master, Start.");
		}

		/*
		 * 获得 Master 数据信息 
		 */
		Master master = this.getMaster();

		/*
		 * 当前有 Master 信息, 且 master 是其他人;
		 */
		if (StringUtils.isNotBlank(master.getNodeID()) && !master.getNodeID().equals(this.nodeID)) {
			if (logger.isDebugEnabled()) {
				logger.debug("other is master, can't try be master, masterid {} myid {}.", master.getNodeID(),
						this.nodeID);
			}
			return;
		}

		/*
		 * 如果我是Mater 节点，则我需要续租;
		 */
		else if (StringUtils.isNotBlank(master.getNodeID())) {
			// do nothing
			if (logger.isDebugEnabled()) {
				logger.debug("I'm master, group id {}.", this.groupID);
			}
		}

		/*
		 * 此过期数据是我认为我是 Master 节点的最长时间
		 */
		long masterLeaseTimeout = leaseTime - this.runTimeAVG;
		long absMasterTimeout = System.currentTimeMillis() + masterLeaseTimeout;
		master.setNodeID(this.nodeID).setAbsTimeoutMS(absMasterTimeout);
		try {
			this.node.propose(this.groupID, master, StateMachine.MasterSmID).get();
		} catch (Throwable e) {
			logger.error("Try Be Master Error,", e);
		}
	}

	@Override
	public int getSmID() {
		return MasterSmID;
	}

	/**
	 * 获得Master 节点信息
	 * 
	 * @return
	 */
	private Master getMaster() {
		return new Master().setNodeID(System.currentTimeMillis() >= this.absExpireTimeMS ? StringUtils.EMPTY // 超过租期，则无
				: masterNodeID) // 返回 masterNodeID
				.setTimeoutMS(this.leaseTimeMS) // 租期
				.setVersion(masterVersion);// 乐观锁
	}

	/**
	 * 此方法处于 Paxos 线程中处理
	 * 
	 * @param nodeID
	 */
	public void checkMasterDropped(String nodeID) {
		if (nodeID.equals(masterNodeID)) {
			this.masterNodeID = null;
			this.eventLoop.execute(beMasterTask);
		}
	}

	/**
	 * 此方法处于提交者线程来处理
	 * 
	 * @return
	 */
	public boolean isIMMaster() {
		return this.nodeID.equals(this.masterNodeID);
	}

	/**
	 * 此方法处于提交者线程来处理
	 * 
	 * @return
	 */
	public boolean noMaster() {
		return StringUtils.isBlank(this.masterNodeID);
	}

	public void needDropMaster() {
		this.eventLoop.execute(() -> {
			this.needDropMaster = true;
		});
	}

	@Override
	public boolean execute(int groupIdx, long instanceID, Value value) {

		/*
		 * Master 数据
		 */
		Master master = value.getEstimateValue();

		/*
		 * 版本不对，不需要处理，防止脑裂
		 */
		if (master.getVersion() != this.masterVersion) {
			return true;
		}

		/*
		 * 设置最新的版本信息: 乐观锁， 设置过期时间
		 */
		this.masterVersion = instanceID;
		this.masterNodeID = master.getNodeID();

		/*
		 * 我认为我是 Master 的时间
		 */
		if (this.masterNodeID.equals(this.nodeID)) {
			this.absExpireTimeMS = master.getAbsTimeoutMS();
		}
		/*
		 * 我认为别人是 Master 的时间
		 */
		else {
			this.absExpireTimeMS = System.currentTimeMillis() + master.getTimeoutMS();
		}

		/*
		 * 打印调试信息
		 */
		if (logger.isDebugEnabled()) {
			logger.debug("Master info masterNodeID : {},  masterVersion: {}, absExpireTimeMS: {}.", this.masterNodeID,
					this.masterVersion, this.absExpireTimeMS);
		}

		return true;
	}
}